package org.bwf.study.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.bwf.study.mapper.OrdersMapper;
import org.bwf.study.model.Orders;
import org.bwf.study.util.JsonUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class OrderConsumer {

    @Autowired
    private OrdersMapper ordersMapper;

    @KafkaListener(topics = "OrderInfo" ,groupId = "myGroup")
    public void ListenGroup(ConsumerRecord<String, String> record, Acknowledgment ack){
        try{
            Orders orders = JsonUtil.decodeJson(record.value(), Orders.class);

            // 数据库认证是否有重复座位

            int result = ordersMapper.insert(orders);
            // 手动提交消息的offerSet
            if(result > 0)
                ack.acknowledge();
        }
        catch(Exception ex){
            log.error("OrderConsumer:" + ex.getMessage());
        }

    }
}
